package com.lry.consumer;

import com.lry.BrokerInfo;
import com.lry.RegisterType;
import com.lry.ServiceThread;
import com.lry.client.RpcClient;
import com.lry.client.RpcProxy;
import com.lry.lock.PutMessageSpinLock;
import com.lry.message.AckMessage;
import com.lry.message.MessageExt;
import com.lry.message.MessageModel;
import com.lry.message.PullMessage;
import com.lry.raft.MqRaftService;
import com.lry.router.BrokerInfoChangeListener;
import com.lry.router.Subscribe;
import com.lry.rpc.MqRpcService;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public class DefaultMqConsumer implements MqConsumer, BrokerInfoChangeListener {

    private MessageModel messageModel;

    private String nameSrv;

    private String group;

    //每个broker子集群中抽出一个出来组成的list
    private List<RpcClient> rpcClients = new ArrayList<>();

    private Map<String,String> subscribeTable = new ConcurrentHashMap<>();

    private ScheduledExecutorService scheduledExecutorService  = Executors.newScheduledThreadPool(5);

    private MessageListener messageListener;

    private String consumer;

    private AckService ackService;

    private AtomicBoolean started = new AtomicBoolean(false);

    private boolean usingRaft;

    private String registerType = RegisterType.ZOOKEEPER;

    private Subscribe subscribe;

    public void setUsingRaft(boolean usingRaft) {
        this.usingRaft = usingRaft;
    }

    public void setRegisterType(String registerType) {
        this.registerType = registerType;
    }

    public DefaultMqConsumer(String nameSrv, String group){
        this(nameSrv,group,UUID.randomUUID().toString());
    }

    public DefaultMqConsumer(String nameSrv,String group,String consumer){
        assert StringUtils.isNotEmpty(nameSrv);
        assert StringUtils.isNotEmpty(group);
        this.nameSrv = nameSrv;
        this.group = group;
        this.consumer = consumer;
        this.ackService = new AckService();
        ackService.start();
    }

    @Override
    public void start() {
        if(usingRaft){
            this.subscribe = new Subscribe(registerType,nameSrv,this);
            subscribe.subscribe();

            //和所有主节点建立连接
            List<BrokerInfo> masters = subscribe.getMasters();
            for (BrokerInfo master : masters) {
                RpcClient rpcClient = new RpcClient(master.getHost(),master.getPort());
                rpcClients.add(rpcClient);
            }
        }else{
            String[] split = this.nameSrv.split(":");
            RpcClient rpcClient = new RpcClient(split[0],Integer.valueOf(split[1]));
            rpcClients.add(rpcClient);
        }
        this.pullTask();
    }

    private void pullTask(){
        scheduledExecutorService.scheduleAtFixedRate(()-> {

            for (RpcClient rpcClient : rpcClients) {

                MqRpcService mqRpcService = RpcProxy.getProxy(rpcClient, MqRpcService.class);
                MqRaftService mqRaftService = RpcProxy.getProxy(rpcClient, MqRaftService.class);

                for (String topic : subscribeTable.keySet()) {
                    String expr = subscribeTable.get(topic);
                    List<MessageExt> messageExts = null;
                    if(usingRaft){
                        messageExts = mqRaftService.pullMessage(new PullMessage(topic,group,consumer,expr));
                    }else{
                        messageExts = mqRpcService.pullMessage(new PullMessage(topic,group,consumer,expr));
                    }
                    if(null==messageExts || messageExts.isEmpty()){
                        continue;
                    }
                    ConsumeStatus consumeStatus = null;
                    if(messageListener instanceof MessageListenerConcurrently){
                        MessageListenerConcurrently messageListenerConcurrently = (MessageListenerConcurrently) messageListener;
                        consumeStatus = messageListenerConcurrently.consumeMessage(messageExts);

                    }else if(messageListener instanceof MessageListenerOrderly){
                        MessageListenerOrderly messageListenerOrderly = (MessageListenerOrderly) messageListener;
                        consumeStatus =  messageListenerOrderly.consumeMessage(messageExts);
                    }

                    //异步ack
                    if(null!=consumeStatus && consumeStatus.equals(ConsumeStatus.CONSUME_SUCCESS)){

                        //针对queueId分组
                        Map<Integer,List<MessageExt>> queueId2Msgs = new HashMap<>();
                        for (MessageExt messageExt : messageExts) {
                            queueId2Msgs.compute(messageExt.getQueueId(),(k,v)->{
                                if(null==v){
                                    v = new ArrayList<>();
                                }
                                v.add(messageExt);
                                return v;
                            });
                        }

                        for (Integer queueId : queueId2Msgs.keySet()) {
                            List<Long> consumeOffsets = new ArrayList<>();
                            for (MessageExt messageExt : queueId2Msgs.get(queueId)) {
                                consumeOffsets.add(messageExt.getQueueOffset());
                            }
                            ackService.putRequest(new AckRequest(topic,group,queueId,consumeOffsets,mqRpcService));
                        }
                    }

                }
            }
        }, 0, 1, TimeUnit.SECONDS);
    }

    @Override
    public void shutdown() {

    }

    @Override
    public void subscribe(String topic, String expr) {
        subscribeTable.put(topic,expr);
    }

    @Override
    public void subscribe(String topic) {
        subscribeTable.put(topic,"");
    }

    @Override
    public void registerMessageListener(MessageListener messageListener) {
        this.messageListener = messageListener;
    }

    public void setMessageModel(MessageModel messageModel) {
        this.messageModel = messageModel;
    }

    @Override
    public void change(BrokerInfo brokerInfo) {

    }

    @AllArgsConstructor
    @NoArgsConstructor
    @Data
    class AckRequest{
        String topic;
        String group;
        int queueId;
        List<Long> consumeOffsets;
        MqRpcService mqRpcService;
    }

    class AckService extends ServiceThread{

        private volatile LinkedList<AckRequest> requestsWrite = new LinkedList<>();
        private volatile LinkedList<AckRequest> requestsRead = new LinkedList<>();
        private final PutMessageSpinLock lock = new PutMessageSpinLock();

        @Override
        public String getServiceName() {
            return AckService.class.getSimpleName();
        }

        @Override
        protected void onWaitEnd() {
            this.swapRequests();
        }

        @Override
        public void run() {
            DefaultMqConsumer.log.info(this.getServiceName() + " service started");

            while (!this.isStopped()) {
                try {
                    this.waitForRunning(10);
                    this.doAck();
                } catch (Exception e) {
                    DefaultMqConsumer.log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }

            // Under normal circumstances shutdown, wait for the arrival of the
            // request, and then flush
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                DefaultMqConsumer.log.warn(this.getServiceName() + " Exception, ", e);
            }

            synchronized (this) {
                this.swapRequests();
            }

            this.doAck();

            DefaultMqConsumer.log.info(this.getServiceName() + " service end");
        }


        private void doAck() {
            if (!this.requestsRead.isEmpty()) {
                for (AckRequest req : this.requestsRead) {
                    req.getMqRpcService().ack(new AckMessage(req.getTopic(),req.getGroup(),req.getQueueId(),req.getConsumeOffsets()));
                }
                this.requestsRead = new LinkedList<>();
            }
        }
        private void swapRequests() {
            lock.lock();
            try {
                LinkedList<AckRequest> tmp = this.requestsWrite;
                this.requestsWrite = this.requestsRead;
                this.requestsRead = tmp;
            } finally {
                lock.unlock();
            }
        }

        public void putRequest(AckRequest request) {
            lock.lock();
            try {
                this.requestsWrite.add(request);
            } finally {
                lock.unlock();
            }
            this.wakeup();
        }


        @Override
        public long getJointime() {
            return 1000 * 60 * 5;
        }
    }
}
